package reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import util.Util;

import java.util.concurrent.*;

/**
 * @author yding5
 */
public class SubscribeOnTest {

    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        Scheduler myScheduler = getMyScheduler();
        Flux.create(sink -> {
            for (int i = 0; i < 10; i++) {
                Util.print("发射了元素" + i);
                sink.next(i);
                Util.sleep(200);
            }
            sink.complete();
        }).doOnRequest(x -> System.out.println("onRequest的线程是：" + Thread.currentThread().getName()))
                .subscribeOn(Schedulers.parallel(), true)
                .publishOn(Schedulers.elastic(), 2)
                .subscribe(
                        // onNext
                        (x) -> Util.print("消费了元素" + x),
                        // onError
                        Throwable::printStackTrace,
                        // onComplete
                        () -> {
                            Util.print("Subscriber consume finished.");
                            countDownLatch.countDown();
                        });
        countDownLatch.await();
        myScheduler.dispose();
    }

    public static Scheduler getMyScheduler() {
        Executor executor = new ThreadPoolExecutor(
                10,  //corePoolSize
                10,  //maximumPoolSize
                0L, TimeUnit.MILLISECONDS, //keepAliveTime, unit
                new LinkedBlockingQueue<>(1000),  //workQueue
                Executors.defaultThreadFactory()
        );
        return Schedulers.fromExecutor(executor);
    }
}
